package org.apache.flink.table.runtime.operators.deduplicate;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;

class DeduplicateFunctionHelper {
    static void processLastRowOnProcTime(RowData currentRow, boolean generateUpdateBefore, boolean generateInsert, ValueState<RowData> state, Collector<RowData> out, boolean isStateTtlEnabled, RecordEqualiser equaliser) throws Exception {
        checkInsertOnly(currentRow);
        if (!generateUpdateBefore && !generateInsert) {
            currentRow.setRowKind(RowKind.UPDATE_AFTER);
            out.collect(currentRow);
        } else {
            RowData preRow = (RowData) state.value();
            state.update(currentRow);
            if (preRow == null) {
                currentRow.setRowKind(RowKind.INSERT);
                out.collect(currentRow);
            } else {
                if (!isStateTtlEnabled && equaliser.equals(preRow, currentRow)) {
                    return;
                }

                if (generateUpdateBefore) {
                    preRow.setRowKind(RowKind.UPDATE_BEFORE);
                    out.collect(preRow);
                }

                currentRow.setRowKind(RowKind.UPDATE_AFTER);
                out.collect(currentRow);
            }
        }

    }

    static void processLastRowOnChangelog(RowData currentRow, boolean generateUpdateBefore, ValueState<RowData> state, Collector<RowData> out, boolean isStateTtlEnabled, RecordEqualiser equaliser) throws Exception {
        RowData preRow = (RowData) state.value();
        RowKind currentKind = currentRow.getRowKind();
        if (currentKind != RowKind.INSERT && currentKind != RowKind.UPDATE_AFTER) {
            if (preRow != null) {
                preRow.setRowKind(RowKind.DELETE);
                out.collect(preRow);
                state.clear();
            } else {
                if (currentKind == RowKind.DELETE) {
                    out.collect(currentRow);
                    state.clear();
                }
            }
        } else {
            if (preRow == null) {
                currentRow.setRowKind(RowKind.INSERT);
                out.collect(currentRow);
            } else {
                if (!isStateTtlEnabled && equaliser.equals(preRow, currentRow)) {
                    return;
                }

                if (generateUpdateBefore) {
                    preRow.setRowKind(RowKind.UPDATE_BEFORE);
                    out.collect(preRow);
                }

                currentRow.setRowKind(RowKind.UPDATE_AFTER);
                out.collect(currentRow);
            }

            currentRow.setRowKind(RowKind.INSERT);
            state.update(currentRow);
        }

    }

    static void processFirstRowOnProcTime(RowData currentRow, ValueState<Boolean> state, Collector<RowData> out) throws Exception {
        checkInsertOnly(currentRow);
        if (state.value() == null) {
            state.update(true);
            out.collect(currentRow);
        }
    }

    static void updateDeduplicateResult(boolean generateUpdateBefore, boolean generateInsert, RowData preRow, RowData currentRow, Collector<RowData> out) {
        if (!generateUpdateBefore && !generateInsert) {
            currentRow.setRowKind(RowKind.UPDATE_AFTER);
            out.collect(currentRow);
        } else if (preRow == null) {
            currentRow.setRowKind(RowKind.INSERT);
            out.collect(currentRow);
        } else {
            if (generateUpdateBefore) {
                RowKind preRowKind = preRow.getRowKind();
                preRow.setRowKind(RowKind.UPDATE_BEFORE);
                out.collect(preRow);
                preRow.setRowKind(preRowKind);
            }

            currentRow.setRowKind(RowKind.UPDATE_AFTER);
            out.collect(currentRow);
        }

    }

    static boolean isDuplicate(RowData preRow, RowData currentRow, int rowtimeIndex, boolean keepLastRow) {
        if (keepLastRow) {
            return preRow == null || getRowtime(preRow, rowtimeIndex) <= getRowtime(currentRow, rowtimeIndex);
        } else {
            return preRow == null || getRowtime(currentRow, rowtimeIndex) < getRowtime(preRow, rowtimeIndex);
        }
    }

    private static long getRowtime(RowData input, int rowtimeIndex) {
        return input.getLong(rowtimeIndex);
    }

    static void checkInsertOnly(RowData currentRow) {
        Preconditions.checkArgument(currentRow.getRowKind() == RowKind.INSERT);
    }

    private DeduplicateFunctionHelper() {
    }
}